package tjs.study.notes.dotest.disruptor.apiLog.config;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Configurable;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Scope;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import tjs.study.notes.dotest.disruptor.parallel.spring.DisruptorNotice;

/**
 * Created by Nicky.Tang on 2022/9/29 上午11:25
 *
 * @since 02.12.10
 */
@Component
@Configurable
@ConditionalOnProperty(name = "shsnc.project.disruptor.enable", matchIfMissing = false, havingValue = "true")
public class DisruptorAutoConfiguration {
    private static final Logger logger = LoggerFactory.getLogger(DisruptorAutoConfiguration.class);
    private DisruptorCompensateThread disruptorCompensateThread;

    @Autowired
    private DisruptorConfig disruptorConfig;

    @Autowired
    @Qualifier("apiLogExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    
    @Bean
    public RingBuffer<DisruptorNotice> ringBuffer() {
        logger.info("【Disruptor初始化】，bufferSize：{}", disruptorConfig.getBufferSize());
        //1 创建RingBuffer
        RingBuffer<DisruptorNotice> ringBuffer = RingBuffer.create(ProducerType.MULTI,
                new EventFactory<DisruptorNotice>() {
                    public DisruptorNotice newInstance() {
                        return new DisruptorNotice();
                    }
                },
                // 预实例化xx个DisruptorNotice（常驻内存中，减少gc影响；缓存行）
                disruptorConfig.getBufferSize(),
                // YieldingWaitStrategy：极低延迟，但是会占用极多CPU
                // SleepingWaitStrategy需要重置下休眠时间，否则默认200ns，仍然会占用较多的cpu
                new SleepingWaitStrategy(200, disruptorConfig.getSleepTimeNs())
        );
        return ringBuffer;
    }


    /**
     * todo-tjs 处理bean销毁时，强制写入逻辑
     *
     * @param ringBuffer
     * @return
     */
    @Bean
    public WorkerPool workerPool(RingBuffer<DisruptorNotice> ringBuffer, ApplicationContext applicationContext) {
        logger.info("【Disruptor初始化】，consumerSize：{}, maxPoolInterval：{}", disruptorConfig.getConsumerSize(), disruptorConfig.getMaxPoolInterval());
        //3 创建含有10个消费者的数组:
        DisruptorAccessKeyConsumer[] disruptorConsumers = new DisruptorAccessKeyConsumer[disruptorConfig.getConsumerSize()];
        for (int i = 0; i < disruptorConsumers.length; i++) {
            disruptorConsumers[i] = applicationContext.getBean(DisruptorAccessKeyConsumer.class, "C" + i);
        }
        //4 构建多消费者工作池
        WorkerPool<DisruptorNotice> workerPool = new WorkerPool<DisruptorNotice>(
                ringBuffer,
                ringBuffer.newBarrier(),
                new EventHandlerException(),
                disruptorConsumers);
        //5 设置多个消费者的sequence序号 用于单独统计消费进度, 并且设置到ringbuffer中
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
        //6 启动workerPool
        workerPool.start(threadPoolTaskExecutor);
        //7 启动补偿线程
        disruptorCompensateThread = new DisruptorCompensateThread(disruptorConfig.getMaxPoolInterval(), disruptorConsumers);
        disruptorCompensateThread.start();
        return workerPool;
    }

    public static class DisruptorCompensateThread extends Thread {
        private static final Logger log = LoggerFactory.getLogger(DisruptorCompensateThread.class);
        private Long maxPoolIntervalSleep;
        private IDisruptorConsume[] disruptorConsumers;

        public DisruptorCompensateThread(Integer maxPoolInterval, IDisruptorConsume[] disruptorConsumers) {
            maxPoolInterval = maxPoolInterval < 5 * 1000 ? 5 * 1000 : maxPoolInterval;
            this.disruptorConsumers = disruptorConsumers;
        }

        @Override
        public void run() {
            log.info("执行Disruptor补偿策略{}。。。", maxPoolIntervalSleep);
            while (true) {
                try {
                    log.trace("执行Disruptor补偿策略{}。。。", maxPoolIntervalSleep);
                    Thread.currentThread().sleep(maxPoolIntervalSleep);
                    for (IDisruptorConsume disruptorConsumer : disruptorConsumers) {
                        if (!disruptorConsumer.isWrite()) {
                            continue;// 这里是会有并发问题，但是牺牲定时的准确性(tryConsume会二次验证)，减少加锁的次数
                        }
                        disruptorConsumer.tryConsume();
                    }
                } catch (Exception ex) {
                    log.error("disruptor补偿机制，运行异常", ex);
                }
            }
        }
    }

    public static class EventHandlerException implements ExceptionHandler {

        private static final Logger log = LoggerFactory.getLogger(EventHandlerException.class);

        @Override
        public void handleEventException(Throwable throwable, long sequence, Object event) {
            throwable.fillInStackTrace();
            log.error("process data error sequence ==[{}] event==[{}] ,ex ==[{}]", sequence, event.toString(), throwable.getMessage());
        }

        @Override
        public void handleOnStartException(Throwable throwable) {
            log.error("start disruptor error ==[{}]!", throwable.getMessage());
        }

        @Override
        public void handleOnShutdownException(Throwable throwable) {
            log.error("shutdown disruptor error ==[{}]!", throwable.getMessage());
        }
    }
}